package org.chnxi.study.messagequeue.rockermq.consumer;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.*;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

public class OrderConsumer {
    public static void main(String[] args) throws MQClientException {
        //创建消费者对象
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer();
        //设置NameServer地址
        consumer.setNamesrvAddr("192.167.50.227:9876;192.167.50.227:9877");
        //设置消费者所在的分组
        consumer.setConsumerGroup("consumerGroup");
        //同一个消费者组内，使用广播模式时，组内所有成员都将受到消息
        //consumer.setMessageModel(MessageModel.CLUSTERING);
        //设置监听的Topic和tag，tag设置为*，表示关注所有tag
        consumer.subscribe("topicName" , "*");
        //设置接受消息处理函数
        consumer.registerMessageListener(new MessageListenerOrderly() {
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext consumeOrderlyContext) {
                for (MessageExt msgExt : list) {
                    System.out.println(Thread.currentThread().getName()+":Message："+new String(msgExt.getBody()));
                }
                //直接返回消费成功
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });
    }
}
